package book.data_clean;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class CleanMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    Data k = new Data();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //解决编码问题
        String line = new String(value.getBytes(), 0, value.getLength(), "GBK");
        //去除数据的第一行也就是列名
        if (line.startsWith("positionName")) return;
        //将去除了列名行的数据以分割符分割
        //因为有些列里面有分割符‘，’
        String[] data = line.split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
        //判断是否有空值，有的话删除
        for (String d : data) {
            if (d == null || d.equals("")) return;
        }
        //判断倒数第二列是否是字母乱码
        //倒数第二列应该是数字，不能有字母、符号、汉字
        Pattern pattern = Pattern.compile("[0-9]*");
        Matcher matcher = pattern.matcher(data[data.length - 2]);
        if (!matcher.matches()) return;
        //将薪资中的9k-10k*2改为9k,10k俩列
//        if (data[1].contains("*")) {
//            String maxS = data[1].split("\\*")[0].split("-")[1];
//            data[1] = data[1].split("-")[0] + "-" + Integer.parseInt(data[1].split("\\*")[1]) *
//                    Integer.parseInt(maxS.substring(0, maxS.length() - 1))+"k";
//        }
        String minSalary = data[1].split("-")[0];
        String maxSalary;
        //如果包含错误符号*
        if (data[1].contains("*")) {
            maxSalary = data[1].split("-")[1].split("\\*")[0];
        } else {
            maxSalary = data[1].split("-")[1];
        }
        //将数据装载如JavaBean
        k.set(data[0], minSalary + "-" + maxSalary, Integer.parseInt(minSalary.substring(0, minSalary.length() - 1)),
                Integer.parseInt(maxSalary.substring(0, maxSalary.length() - 1)), data[2], data[3], data[4], data[5], data[6], data[7], data[8], data[9], Integer.parseInt(data[10]), Integer.parseInt(data[11]), data[12]);
        //将数据输出
        context.write(new Text(k.toString()), NullWritable.get());
    }
}
